#pragma once

#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>

#if defined(__clang__) || defined(__GNUC__)
#define CPP_STANDARD __cplusplus
#elif defined(_MSC_VER)
#define CPP_STANDARD _MSVC_LANG
#endif

#if CPP_STANDARD >= 201703L
#define HAS_CPP_17
#include <optional>
#endif

#include <cassert>

namespace ccq {

class ThreadPool {
public:
    using task_t = std::function<void()>;
#ifdef HAS_CPP_17
    template <typename F, typename... Args>
    using result_of_t = std::invoke_result_t<F, Args...>;
#else
    template <typename F, typename... Args>
    using result_of_t = typename std::result_of<F(Args...)>::type;
#endif

    explicit ThreadPool(unsigned int num_threads = std::thread::hardware_concurrency());
    ~ThreadPool();

    static ThreadPool& globalThreadPool() {
        static ThreadPool _pool;
        return _pool;
    }

    template <typename F, typename... Args>
    auto post(F &&f, Args &&... args) -> std::future<result_of_t<F, Args...>>;

    void stop();
    void clear();

    bool isStopped();
    void setThreadCount(unsigned int num_threads);
    unsigned int threadCount() const;
    unsigned int taskCount() const;

private:
    void addThread();

private:
    std::vector<std::thread> m_threads;
    std::queue<task_t> m_tasks;
    mutable std::mutex m_queue_mutex;
    std::condition_variable m_condition;
    bool m_stopped = false;
};

static ThreadPool& globalThreadPool() {
    return ThreadPool::globalThreadPool();
}

namespace concurrent {
template <typename F, typename... Args>
static auto async(F&& f, Args&&... args) {
    return globalThreadPool().post(std::forward<F>(f), std::forward<Args>(args)...);
}
} // namespace concurrent

inline ThreadPool::ThreadPool(unsigned int num_threads) {
    setThreadCount(num_threads);
}

inline ThreadPool::~ThreadPool() {
    stop();
}

inline  unsigned int ThreadPool::threadCount() const {
    return m_threads.size();
}

inline unsigned int ThreadPool::taskCount() const {
    std::lock_guard<std::mutex> lock(m_queue_mutex);
    return m_tasks.size();
}

template <typename F, typename... Args>
inline auto ThreadPool::post(F &&f, Args &&... args) -> std::future<result_of_t<F, Args...>> {
    using result_t = result_of_t<F, Args...>;
    auto task = std::make_shared<std::packaged_task<result_t()>>(std::bind(std::forward<F>(f), std::forward<Args>(args)...));
    std::future<result_t> result = task->get_future();
    {
        std::unique_lock<std::mutex> lock(m_queue_mutex);
        if (m_stopped) {
            throw std::runtime_error("ThreadPool is stopped");
        }
        m_tasks.emplace([task]() { (*task)(); });
    }
    m_condition.notify_one();
    return result;
}

inline void ThreadPool::stop() {
    {
        std::unique_lock<std::mutex> lock(m_queue_mutex);
        m_stopped = true;
    }
    m_condition.notify_all();
    for (auto& thread : m_threads) {
        if (thread.joinable()) {
            thread.join();
        }
    }
    m_threads.clear();
}

inline void ThreadPool::clear() {
    std::unique_lock<std::mutex> lock(m_queue_mutex);
    std::queue<task_t> t;
    m_tasks.swap(t);
}

inline bool ThreadPool::isStopped()
{
    std::unique_lock<std::mutex> lock(m_queue_mutex);
    return m_stopped;
}

inline void ThreadPool::setThreadCount(unsigned int num_threads)
{
    assert(num_threads > 0);
    std::unique_lock<std::mutex> lock(m_queue_mutex);
    if(m_stopped)
        return;
    if(m_threads.size() == num_threads) {
        return;
    }
    else if(m_threads.size() < num_threads) {
        num_threads -=  m_threads.size();
        for (unsigned int i = 0; i < num_threads; ++i) {
            addThread();
        }
    }
    else {
        // FIXME: Not support reduce the number of active threads
    }

}

inline void ThreadPool::addThread()
{
    m_threads.emplace_back([this]() {
        while (true) {
            std::unique_lock<std::mutex> lock(m_queue_mutex);
            m_condition.wait(lock, [this]() { return m_stopped || !m_tasks.empty(); });
            if (m_stopped && m_tasks.empty()) {
                return;
            }
            task_t task = std::move(m_tasks.front());
            m_tasks.pop();
            lock.unlock();
            task(); // execute task outside lock
        } });
}


}   // namespace ccq
